package com.unism.uniMeterage.meterage.stream.version1_0;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;

public abstract class AbstractWork implements WorkNode {
    private ConcurrentLinkedQueue<Object> queue;
    private WorkNodeManager parent;
    private Thread myThread;
    private String name;
    private static Integer dataNumber;
    private static Integer Default_Num=1;
    public AbstractWork(WorkNodeManager parent, String name,Integer number) {
        this.queue = new ConcurrentLinkedQueue<Object>();
        this.parent = parent;
        this.myThread = null;
        this.name = name;
        if(number == null || number < 1){
            dataNumber=Default_Num;
        }else {
            dataNumber=number;
        }
        this.init();
    }

    @Override
    public void init() {
      System.out.println("当前工作节点["+this.name+"]开始初始化");
      if(queue != null && parent !=null && name !=null){
          if(this.initService()){
              this.destory();
          }
      }else{
          this.destory();
      }
    }
    /**
     * 私有节点内部初始化服务方法
     */
    private Boolean initService(){
       try{
           myThread=new Thread(this.name){
               @Override
               public void run() {
                   while (true){
                     List<Object> data =subscribe(dataNumber);
                     if(data !=null && !data.isEmpty()){
                         if(execute(data)){
                             success(data);
                         }else{
                             failure(data);
                         }
                     }else{
                        // System.out.println("当前工作节点["+name+"]没有订阅到有效数据");
                     }
                   }
               }
           };
           myThread.start();
           return  true;
       }catch(Exception e){
           e.printStackTrace();
           return  false;
       }
    }
    @Override
    public void destory() {
        System.out.println("当前工作节点["+this.name+"]被销毁");
    }
    @Override
    public void pubish(String name,Object data) {
       System.out.println("消息发送成功。。。");
    }
    @Override
    public List<Object> subscribe(Integer num) {
        if(!this.queue.isEmpty()){
            List<Object> result=new ArrayList<Object>();
            int index=0;
            while (!this.queue.isEmpty() && index <=num){
                Object obj=this.queue.poll();
                if(obj !=null){
                    result.add(obj);
                }
            }
            return  result;
        }
        return null;
    }
    @Override
    public void failure(Object obj) {

    }

    @Override
    public void success(Object obj) {

    }
    @Override
    public String getName() {
        return this.name;
    }

    @Override
    public Thread currThread() {
        return this.myThread;
    }

    @Override
    public ConcurrentLinkedQueue currQueue() {
        return this.queue;
    }

    @Override
    public Integer currStatus() {
        return null;
    }
}

